package com.iqiyi.pps.epg.core.service.program;

import com.iqiyi.kiwi.utils.Configuration;
import com.iqiyi.pps.epg.api.model.web.query.TFilter;
import com.iqiyi.pps.epg.core.async.AsyncService;
import com.iqiyi.pps.epg.core.async.HeartBeatCenter;
import com.iqiyi.pps.epg.core.async.IAsyncTask;
import com.iqiyi.pps.epg.core.async.IHeartBeat;
import com.iqiyi.pps.epg.core.model.channel.ChannelConfig;
import com.iqiyi.pps.epg.core.model.program.ProgramData;
import com.iqiyi.pps.epg.core.model.program.ProgramLog;
import com.iqiyi.pps.epg.core.model.rtmp.RtmpServer;
import com.iqiyi.pps.epg.core.model.stream.LiveChannelStream;
import com.iqiyi.pps.epg.core.model.task.ProgramSimpleTask;
import com.iqiyi.pps.epg.core.model.task.ProgramSimpleTaskType;
import com.iqiyi.pps.epg.core.model.task.SimpleTaskStatus;
import com.iqiyi.pps.epg.core.service.channel.ChannelService;
import com.iqiyi.pps.epg.core.service.rtmp.RtmpService;
import com.iqiyi.pps.epg.core.service.stream.LiveChannelStreamService;
import com.iqiyi.pps.epg.core.service.task.ProgramSimpleTaskService;
import com.iqiyi.pps.epg.core.utils.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.util.*;


/**
 * Created by karkarrotxu on 2015/12/15.
 */
@Service
@Lazy(false)
public class SnapAsyncServiceImpl extends AsyncService implements InitializingBean, IHeartBeat, ISnapAsyncService {

    private static final Logger logger = LoggerFactory.getLogger(SnapAsyncServiceImpl.class);
    private final static int HEARTBEAT_DELAY = 30 * 1000; //
    public static final String SNAP_STREAM_TYPE = "SNAP_STREAM_TYPE";
    private long heartBeatTime = 0;
    public static String ASSIGN_WORK_IP = "10.153.132.20";
    public static final int SNAPSHOT_PREVIEW_SECOND = Configuration.getInt("snapshot.preview.second");
    public static final int DEFAULT_SNAPSHOT_GAP_SECOND = Configuration.getInt("snapshot.gap.second");
    //rtmp://111.206.22.100/live/ls_susa1027_720p
    public static final String FORMAT_RTMP_URL = "rtmp://%s/live/%s";
    public static final String DEFAULT_STREAM_TYPE = "HIGH_DEFINITION";
    @Autowired
    private ProgramSimpleTaskService simpleTaskService;
    @Autowired
    private ProgramService programService;
    @Autowired
    private LiveChannelStreamService liveChannelStreamService;
    @Autowired
    private RtmpService rtmpService;
    @Autowired
    private ChannelService channelService;

    private long lastUpdateRtmpTime = 0;
    private Map<String, String> transIpMap = new HashMap<>(50);

    public void afterPropertiesSet() throws Exception {
        logger.info("SnapAsyncServiceImpl afterPropertiesSet");
        HeartBeatCenter.add(this);
        tick(System.currentTimeMillis());
    }

    protected boolean queryEnable(){
        return getEnable("snap");
    }

    @Override
    public void tick(long now) {
        if (now - heartBeatTime >= HEARTBEAT_DELAY) {
            heartBeatTime = now;
            logger.info("triggerOnce tick");
            triggerStopOnce();
            triggerStartOnce();
        }
    }


    @Override
    public void addProgramTask(ProgramData program) {
        interruptTask(program.getId());
        if (program.getSnapShotType() == 1 && program.getDeleteFlag() == 0) {
            if (program.getStartPlayTime().getTime() >= program.getStopPlayTime().getTime()) {
                logger.error("addProgramTask return [pid={}][start={}][end={}]", program.getId(), program.getStartPlayTime().getTime(), program.getStopPlayTime().getTime());
                return;
            }
            if (program.getStopPlayTime().getTime() < System.currentTimeMillis()) {
                logger.info("addProgramTask skip [pid={}][start={}][end={}]", program.getId(), program.getStartPlayTime().getTime(), program.getStopPlayTime().getTime());
                return;
            }
            addSnapStartTask(program);
            // addSnapStopTask(program);
        }
    }

    @Override
    public void addStreamTask(LiveChannelStream liveChannelStream) {
        String streamType = getAssignStreamType();
        if (null != streamType && streamType.equals(liveChannelStream.getStreamType())) {
            String currentTime = DateTimeUtil.transLongFormat(System.currentTimeMillis());
            logger.info("addStreamTask[cid={}][streamType={}][lcsId={}]", liveChannelStream.getFatherId(), streamType, liveChannelStream.getId());
            Map<String, String> params = new HashMap<>();
            params.put("EQI_deleteFlag", 0 + "");
            params.put("EQI_fatherId", liveChannelStream.getFatherId() + "");
            params.put("GET_stopPlayTime", currentTime);
            params.put("EQI_snapShotType", 1 + "");
            TFilter tFilter = new TFilter();
            tFilter.setFilters(params);
            List<ProgramData> programDataList = programService.findProgramDataList(tFilter, "startPlayTime", "asc");
            logger.info("[addStreamTask][cid={}][size={}]", liveChannelStream.getFatherId(), programDataList.size());
            if (programDataList.size() > 0) {
                for (ProgramData programData : programDataList) {
                    addProgramTask(programData);
                }
            }
        }
    }

    private void addSnapStartTask(ProgramData program) {
        int gapTime = program.getSnapShotGap();
        if ( gapTime < 0) {
            gapTime = DEFAULT_SNAPSHOT_GAP_SECOND;
        }
        ProgramSimpleTask programSimpleTask = new ProgramSimpleTask();
        programSimpleTask.setServerIp(RequestUtils.getLoadlIpByOS());
        programSimpleTask.setWorkIp(isProductionEnv() ? ASSIGN_WORK_IP : programSimpleTask.getServerIp());
        programSimpleTask.setTaskType(ProgramSimpleTaskType.NOTIFY_SNAP_START.getTypeIndex());
        programSimpleTask.setEntityId(program.getId());
        programSimpleTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        programSimpleTask.setStartUpTime(new Timestamp(program.getStartPlayTime().getTime() - (gapTime + SNAPSHOT_PREVIEW_SECOND) * DateTimeUtil.SECOND));
        programSimpleTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        programSimpleTask.setExpireTime(new Timestamp(program.getStopPlayTime().getTime()));
        simpleTaskService.saveTask(programSimpleTask);
        triggerStartOnce();
    }

    public void triggerStartOnce() {
        logger.info("NOTIFY_SNAP_START tick");
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                logger.info("NOTIFY_SNAP_START query");
                executeSimpleTaskList(ProgramSimpleTaskType.NOTIFY_SNAP_START.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    public void triggerStopOnce() {
        logger.info("NOTIFY_SNAP_STOP tick");
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                logger.info("NOTIFY_SNAP_STOP query");
                executeSimpleTaskList(ProgramSimpleTaskType.NOTIFY_SNAP_STOP.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    private void addSnapStopTask(ProgramData program) {
        ProgramSimpleTask programSimpleTask = new ProgramSimpleTask();
        programSimpleTask.setServerIp(RequestUtils.getLoadlIpByOS());
        programSimpleTask.setWorkIp(isProductionEnv() ? ASSIGN_WORK_IP : programSimpleTask.getServerIp());
        programSimpleTask.setTaskType(ProgramSimpleTaskType.NOTIFY_SNAP_STOP.getTypeIndex());
        programSimpleTask.setEntityId(program.getId());
        programSimpleTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        programSimpleTask.setStartUpTime(new Timestamp(program.getStopPlayTime().getTime()));
        programSimpleTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        programSimpleTask.setExpireTime(new Timestamp(program.getStopPlayTime().getTime() + DateTimeUtil.DAY_TIME));
        simpleTaskService.saveTask(programSimpleTask);
        triggerStopOnce();
    }

    @Override
    public void interruptTask(long programId) {
        int effectNum = simpleTaskService.interruptTask(programId, ProgramSimpleTaskType.NOTIFY_SNAP_START.getTypeIndex());
        logger.info("[effectNum]NOTIFY_SNAP_START pId {},effectNum={}:", programId, effectNum);
        effectNum = simpleTaskService.interruptTask(programId, ProgramSimpleTaskType.NOTIFY_SNAP_STOP.getTypeIndex());
        logger.info("[effectNum]NOTIFY_SNAP_STOP pId {},effectNum={}:", programId, effectNum);
        if (effectNum > 0) {
            String result = ApiSnapshotUtil.doSnapStop(programId);
            logger.info("[doSnapStop][programId={}][result={}]:", programId, result);
            ProgramLog programLog = WriteSystemLogUtil.getInstance().getProgramLog(programId, Constants.LogType.STREAM_SNAPSHOT_INTERRUPT.getIndex(), " notify stop: " + result);
            programService.saveProgramLog(programLog);
        }
    }

    @Override
    public void interruptTaskByChannel(long channelId) {
        String currentTime = DateTimeUtil.transLongFormat(System.currentTimeMillis());
        Map<String, String> params = new HashMap<>();
        params.put("EQI_deleteFlag", 0 + "");
        params.put("EQI_fatherId", channelId + "");
        params.put("GET_stopPlayTime", currentTime);
        params.put("EQI_snapShotType", 1 + "");
        TFilter tFilter = new TFilter();
        tFilter.setFilters(params);
        List<ProgramData> programDataList = programService.findProgramDataList(tFilter, "startPlayTime", "asc");
        logger.info("[interruptTaskByChannel][size={}]", programDataList.size());
        if (programDataList.size() > 0) {
            for (ProgramData programData : programDataList) {
                interruptTask(programData.getId());
            }
        }
    }


    protected void executeSimpleTaskList(int type, int status, boolean needOverride) {
        logger.info("[executeSimpleTaskList] execute type:{},status={}", type, status);
        try {
            List<ProgramSimpleTask> resultList = simpleTaskService.findList(type, status);
            if (resultList == null || resultList.size() == 0) {
                logger.info("[result empty] resultList == null :" + (resultList == null));
                return;
            }
            logger.info("[executeSimpleTaskList] execute size:" + resultList.size());
            long currentTime = System.currentTimeMillis();
            if (needOverride) {
                Collections.reverse(resultList);
            }
            List<ProgramSimpleTask> updateList = new ArrayList<>(2048);
            Map<Long, ProgramSimpleTask> checkMap = new HashMap<>(2048);
            List<ProgramSimpleTask> waitToDealList = new ArrayList<>(2048);
            for (ProgramSimpleTask task : resultList) {
                if (currentTime > task.getExpireTime().getTime()) {
                    task.setTaskStatus(SimpleTaskStatus.EXPIRED.getStatusIndex());
                    updateList.add(task);
                } else {
                    if (needOverride) {
                        if (!checkMap.containsKey(task.getEntityId())) {
                            checkMap.put(task.getEntityId(), task);
                        } else {
                            task.setTaskStatus(SimpleTaskStatus.OVERRIDE.getStatusIndex());
                            updateList.add(task);
                        }
                    } else {
                        waitToDealList.add(task);
                    }
                }
            }
            logger.info("[executeSimpleTaskList] overtime size:" + updateList);
            if (updateList.size() > 0) {
                simpleTaskService.batchUpdateList(updateList);
            }

            if (checkMap.size() > 0) {
                waitToDealList.addAll(checkMap.values());
            }

            if (waitToDealList.size() > 0) {
                updateList.clear();
                for (ProgramSimpleTask task : waitToDealList) {
                    if (executeSimpleTask(task)) {
                        updateList.add(task);
                    }
                }
                if (updateList.size() > 0) {
                    simpleTaskService.batchUpdateList(updateList);
                }
            }
        } catch (Exception e) {
            logger.error("[executeSimpleTaskList] error", e);
        }
        logger.info("[executeSimpleTaskList] execute end");
    }

    protected boolean executeSimpleTask(ProgramSimpleTask simpleTask) {
        ProgramSimpleTaskType type = ProgramSimpleTaskType.getTypeByIndex(simpleTask.getTaskType());
        boolean flag = false;
        switch (type) {
            case NOTIFY_SNAP_START:
                flag = doSnapStart(simpleTask);
                break;
            case NOTIFY_SNAP_STOP:
                flag = doSnapStop(simpleTask);
                break;
        }
        if (flag) {
            simpleTask.setTaskStatus(SimpleTaskStatus.COMPLETE.getStatusIndex());
            simpleTask.setCompleteTime(new Timestamp(System.currentTimeMillis()));
        }
        return flag;
    }

    private String getAssignStreamType() {
        List<ChannelConfig> channelConfigList = channelService.getConfig(SNAP_STREAM_TYPE);
        if (channelConfigList.size() == 0) {
            return DEFAULT_STREAM_TYPE;
        } else {
            return channelConfigList.get(0).getVal();
        }
    }

    private boolean doSnapStart(ProgramSimpleTask simpleTask) {
        ProgramData programData = programService.findProgramDataById(simpleTask.getEntityId());
        ProgramLog programLog = WriteSystemLogUtil.getInstance().getProgramLog(simpleTask.getEntityId(), Constants.LogType.STREAM_SNAPSHOT_START.getIndex(), "");
        if (programData == null || programData.getSnapShotType() != 1) {
            logger.info("[doSnapStart]skip [{sid={}][pid={}]", simpleTask.getId(), simpleTask.getEntityId());
            programLog.setDescription(" skip task case : [programData is null=" + (programData == null) + "][snapshotType=" + programData.getSnapShotType() + "]");
            programService.saveProgramLog(programLog);
            return true;
        }
        String snapStreamType = getAssignStreamType();
        LiveChannelStream liveChannelStream = liveChannelStreamService.getByStreamTypeWithIsEnable(Long.valueOf(programData.getFatherId()), snapStreamType, true);
        if (liveChannelStream == null) {
            logger.info("[has no such liveChannelStream[{taskId={}][pid={}][snapStreamType={}]", simpleTask.getId(), simpleTask.getEntityId(), snapStreamType);
            programLog.setDescription("can not find snapStreamType=" + snapStreamType);
            programService.saveProgramLog(programLog);
            return false;
        }
        String privateIp = getTransIpMap().get(liveChannelStream.getStreamIp());
        if (privateIp == null) {
            logger.info("[has no such privateIp][streamId={}][publicIp={}]", liveChannelStream.getId(), liveChannelStream.getStreamIp());
            programLog.setDescription("[lcs Id =" + liveChannelStream.getId() + "] can not find privateIp  , publicIp=" + liveChannelStream.getStreamIp());
            programService.saveProgramLog(programLog);
            return false;
        }
        String rtmpurl = String.format(FORMAT_RTMP_URL, privateIp, liveChannelStream.getStreamName());
        String result = ApiSnapshotUtil.doSnapStart(programData.getId(), rtmpurl,programData.getSnapShotGap());
        programLog.setDescription("[streamName:" + liveChannelStream.getStreamName() + "][result=" + result + "]");
        programService.saveProgramLog(programLog);
        boolean suc = ApiSnapshotUtil.STR_CODE_SUC.equals(result);
        if (suc) {
            addSnapStopTask(programData);
        }
        return suc;
    }

    private Map<String, String> getTransIpMap() {
        long current = System.currentTimeMillis();
        if (current > lastUpdateRtmpTime + DateTimeUtil.MINUTE_TIME) {
            lastUpdateRtmpTime = current;
            transIpMap.clear();
            List<RtmpServer> serverList = rtmpService.getServerList();
            for (RtmpServer server : serverList) {
                transIpMap.put(server.getPublicIp(), server.getPrivateIp());
            }
        }
        return transIpMap;
    }

    private boolean doSnapStop(ProgramSimpleTask simpleTask) {
        ProgramLog programLog = WriteSystemLogUtil.getInstance().getProgramLog(simpleTask.getEntityId(), Constants.LogType.STREAM_SNAPSHOT_STOP.getIndex(), "");
        String result = ApiSnapshotUtil.doSnapStop(simpleTask.getEntityId());
        programLog.setDescription(result);
        programService.saveProgramLog(programLog);
        boolean suc = ApiSnapshotUtil.STR_CODE_SUC.equals(result);
        return suc;
    }
}
